"""
Case Type   : 极致RTO支持备机可读
Case Name   : 资源池化存储复制双集群，主集群创建普通行存段页式表关联物化视图，备集群开启极致RTO场景下验证备机视图可读
Create At   : 2025/2
Owner       : haomeng
Description :
    1、备集群内开启极致RTO特性，重启使其生效
    2、主集群内创建表关联物化视图
    3、执行tpcc读写业务
    4、刷新物化视图，稍后备备集群首备查看视图数据
    5、对比主备机视图数据一致性
    6、备集群从备查看视图数据
    7、清理环境，恢复环境初始状态
Expect      :
    1、设置成功，重启成功
    2、主集群创建物化视图成功
    3、主集群读写业务成功
    4、备集群刷新视图成功，查看视图数据成功
    5、主备集群视图数据一致
    6、备集群从备查看视图数据失败
    7、清理成功，恢复环境成功
History     :
"""

import os
import time
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class RTOStandbyRead(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.root_user = Node(node='PrimaryRoot')
        self.p_pri_sh = CommonSH('P_PrimaryDbUser')
        self.s_pri_sh = CommonSH('S_PrimaryDbUser')
        self.s_sta_sh = CommonSH('S_Standby1DbUser')
        self.constant = Constant()
        self.com = Common()

        self.param_list = ['recovery_parse_workers', 'recovery_redo_workers', 'ss_disaster_mode']
        self.value_list = ['4', '4']
        self.default_value = []
        self.table_list = ['bmsql_config', 'bmsql_customer', 'bmsql_district', 'bmsql_history',
                           'bmsql_item', 'bmsql_new_order', 'bmsql_oorder', 'bmsql_order_line',
                           'bmsql_stock', 'bmsql_warehouse']
        self.mview_list = ['bmsql_config_mview', 'bmsql_customer_mview', 'bmsql_district_mview', 'bmsql_history_mview',
                          'bmsql_item_mview', 'bmsql_new_order_mview', 'bmsql_oorder_mview', 'bmsql_order_line_mview',
                          'bmsql_stock_mview', 'bmsql_warehouse_mview']

        self.log.info('---校验当前环境是否为存储复制双集群---')
        mode = self.p_pri_sh.execut_db_sql(f'show {self.param_list[2]};')
        self.log.info(mode)
        self.assertIn('dorado', mode, '当前环境不是存储复制双集群')

    def test_rto_standby_read(self):
        text = '-----step1:备集群内开启极致RTO特性，重启使其生效;  expect:设置成功，重启成功-----'
        self.log.info(text)

        for param in self.param_list[:2]:
            value = self.s_pri_sh.execut_db_sql(f'show {param}')
            self.default_value.append(value.splitlines()[-2].strip())
        self.log.info(self.default_value)

        for param in self.param_list[:2]:
            set_param = self.s_pri_sh.execute_gsguc('set',
                        self.constant.GSGUC_SUCCESS_MSG, f'{param} = {self.value_list[self.param_list.index(param)]}')
            self.log.info(set_param)
            self.assertTrue(set_param, '参数设置失败')

        self.log.info('---重启集群使参数生效---')
        restart = self.s_pri_sh.restart_db_cluster()
        self.log.info(restart)
        self.assertTrue(restart, '集群重启失败')

        self.log.info('---查看参数值修改是否成功---')
        for param in self.param_list[:2]:
            value = self.s_pri_sh.execut_db_sql(f'show {param};')
            self.assertEqual(value.splitlines()[-2].strip(),
                             self.value_list[self.param_list.index(param)], '参数未修改成功')

        text = '-----step2:创建表关联物化视图  expect:创建视图成功-----'
        self.log.info(text)

        create_mview_sql = f"create materialized view {self.mview_list[0]} as select * from {self.table_list[0]};"
        for i in range(1, len(self.mview_list)):
            create_mview_sql += f"create materialized view {self.mview_list[i]} as select * from {self.table_list[i]};"
        self.log.info(create_mview_sql)
        create_res = self.p_pri_sh.execut_db_sql(create_mview_sql)
        self.log.info(create_res)
        self.assertEqual(10, create_res.count(self.constant.CREATE_MATERIALIZED_VIEW_MSG), '创建物化视图失败')

        text = '-----step3:主节点执行tpcc读写业务  expect:业务执行成功-----'
        self.log.info(text)

        tpcc_cmd = f'./runBenchmark.sh {macro.PROPS_FILENAME}'
        self.log.info(tpcc_cmd)

        tpcc_execute = self.com.start_tpcc(self.root_user, f'{macro.TPCC_PATH}', tpcc_cmd)
        self.log.info(tpcc_execute)
        self.assertIn('Session End', tpcc_execute, 'tpcc执行失败')

        text = '-----step4:刷新物化视图且查看视图数据  expect:刷新成功&查看成功-----'
        self.log.info(text)

        sta_res_list = []
        for view in self.mview_list:
            refresh_sql = f"refresh materialized view {view};"
            self.log.info(refresh_sql)
            refresh_res = self.p_pri_sh.execut_db_sql(refresh_sql)
            self.log.info(refresh_res)

            sta_sql = f"select count(*) from {view};"
            self.log.info(sta_sql)
            sta_res = self.s_pri_sh.execut_db_sql(sta_sql)
            self.log.info(sta_res)
            sta_res_list.append(sta_res)
            self.assertNotIn('(0 rows)', sta_res, '查询结果异常')

        text = '-----step5:对比主备集群物化视图数据一致性  expect:查看成功，与主集群数据一致-----'
        self.log.info(text)

        pri_res_list = []
        for view in self.mview_list:
            sql = f"select count(*) from {view};"
            self.log.info(sql)
            pri_res = self.p_pri_sh.execut_db_sql(sql)
            self.log.info(pri_res)
            pri_res_list.append(pri_res)
            self.assertNotIn('(0 rows)', pri_res, '查询结果异常')

        time.sleep(5)
        self.assertEqual(pri_res_list, sta_res_list, '主备查询结果不一致')

        text = '-----step6:备集群从备节点查看视图数据  expect:查看失败-----'
        self.log.info(text)

        ss_select = self.s_sta_sh.execut_db_sql(f"select count(*) from {self.view_list[0]};")
        self.log.info(ss_select)
        self.assertIn(self.constant.disaster_read_fail_msg, ss_select, '从备节点查询成功')

    def tearDown(self):
        text = '-----step7:清理环境，恢复环境初始状态  expect:清理成功，恢复环境成功-----'
        self.log.info(text)

        drop_sql = f"drop materialized view {self.mview_list[0]};"
        for i in range(1, len(self.mview_list)):
            drop_sql += f"drop materialized view {self.mview_list[i]};"
        self.log.info(drop_sql)
        drop_res = self.p_pri_sh.execut_db_sql(drop_sql)
        self.log.info(drop_res)

        for param in self.param_list[:2]:
            reset_param = self.s_pri_sh.execute_gsguc('set',
                                                      self.constant.GSGUC_SUCCESS_MSG,
                                                      f'{param} = {self.default_value[self.param_list.index(param)]}')
            self.log.info(reset_param)

        restart = self.s_pri_sh.restart_db_cluster(False)
        self.log.info(restart)

        for param in self.param_list[:2]:
            reset_value = self.s_pri_sh.execut_db_sql(f'show {param};')
            self.log.info(reset_value)
            self.assertEqual(reset_value.splitlines()[-2].strip(),
                             self.default_value[self.param_list.index(param)], '参数恢复失败')
        self.assertEqual(drop_res.count(self.constant.DROP_MATERIALIZED_VIEW_MSG),
                         10, '删除物化视图失败')
        self.log.info(f'----{os.path.basename(__file__)} end----')
